package SparkExercises.kejian5.lianxi1;

import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.Function2;
import org.apache.spark.api.java.function.PairFunction;
import scala.Tuple2;

import java.util.List;

public class LianX {
    public static void main(String[] args) {
        /**
         * 练习：有一份新能源车实时上报的车辆动态数据
         * （字段说明：vin：车架号，是车辆的唯一表示，
         * collect_time：采集时间，是数据的上传时间，
         * speed：车速，是当前采集时间所对应的车辆速度），
         * 请写一个spark程序，统计出每辆车每天的最大车速，并将信息输出到控制台，如下图所示：
         */
        SparkConf sparkConf = new SparkConf().setAppName("LianXi").setMaster("local");
        JavaSparkContext sc = new JavaSparkContext(sparkConf);

        Gson gson1 = new GsonBuilder().create();

        JavaRDD<String> rdd1 = sc.textFile("./data/can_data.json");

        JavaPairRDD<String, Car> rdd2 = rdd1.mapToPair(new PairFunction<String, String, Car>() {
            @Override
            public Tuple2<String, Car> call(String s) throws Exception {
                Car car = new Gson().fromJson(s, Car.class);
                return new Tuple2<>(car.getVin()+"_"+car.getCollectTime().split(" ")[0],car);
            }
        });

        JavaPairRDD<String, Car> rdd3 = rdd2.reduceByKey(new Function2<Car, Car, Car>() {
            @Override
            public Car call(Car car, Car car2) throws Exception {
                if (car.getSpeed() > car2.getSpeed()) {
                    return car;
                } else {
                    return car2;
                }
//                return null;
            }
        });
        System.out.println(rdd2.collect());
//        JavaRDD<Car> values = rdd3.values();
//        List<Car> collect = values.collect();
//        for (Car c: collect){
//            System.out.println(c);
//        }
    }
}



